package com.wenx.activeMqCamel;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.camel.component.ActiveMQComponent;
import org.apache.activemq.jms.pool.PooledConnectionFactory;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.jms.JmsComponent;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.impl.DefaultMessage;

import javax.jms.ConnectionFactory;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;

/**
 * 连接池连接设置属性
 */
public class activeMQProvider3 {
    private static String user = "admin";
    private static String password = "admin";
    //填写AMQ的IP
    private static String url = "failover://tcp://10.10.3.186:61616";
    public static void main(String[] args) throws Exception {
        CamelContext context = new DefaultCamelContext(); // 1. 创建 CamelContext.
        //创建AMQ的连接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
            //自定义连接池
//        PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory();
//        pooledConnectionFactory.setConnectionFactory(connectionFactory);
//
//        pooledConnectionFactory.setMaximumActiveSessionPerConnection(200);
//        pooledConnectionFactory.setIdleTimeout(120);
//        pooledConnectionFactory.setMaxConnections(50);
//        pooledConnectionFactory.setBlockIfSessionPoolIsFull(true);
//        pooledConnectionFactory.start();

        ActiveMQComponent activeMQComponent = new ActiveMQComponent();
        activeMQComponent.setConnectionFactory(connectionFactory);
        //使用连接池连接
        activeMQComponent.setUsePooledConnection(true);
        //事务
        activeMQComponent.setTransacted(true);
        //客户端确认
        activeMQComponent.setAcknowledgementMode(3);
        //持久化
        activeMQComponent.setDeliveryPersistent(true);
//        //clientid
//        activeMQComponent.setClientId("aaa");
//        //topic只有在持久订阅（durablesubscription）下是持久化的
//        activeMQComponent.setDurableSubscriptionName("durable");
        context.addComponent("activemq", activeMQComponent);
        System.out.println(url + " " + user + " " + password);
        context.setTracing(true);
        context.start(); // 4. 启动CamelContext.
        System.out.println("状态:"+context.getStatus().toString());
        context.addRoutes((new activeMQProvider3()).new test1());
        Thread.sleep(Integer.MAX_VALUE);  // 为了保持CamelContext处于工作状态，这里需要sleep主线程
        context.stop(); // 最后停止CamelContext
    }
    public class test1 extends RouteBuilder{
        @Override
        public void configure() throws Exception {
            from("timer:foo?repeatCount=1")
                    .to("http://localhost:8080/demo/hello6?id=123&name=abd")
                    //.setHeader("type",jsonPathExpression)
                    .process(new Processor() {
                        @Override
                        public void process(Exchange exchange) throws Exception {
                            //解析exchange中的返回值
                            DefaultMessage message = (DefaultMessage) exchange.getIn();
                            InputStream bodyStream =  (InputStream)message.getBody();
                            ByteArrayOutputStream results = new ByteArrayOutputStream();
                            byte[] buffer = new byte[1024];
                            int length;
                            while ((length = bodyStream.read(buffer)) != -1) {
                                results.write(buffer, 0, length);
                            }
                            //rest响应数据集
                            String body = results.toString("UTF-8");
                            results.close();
                            bodyStream.close();
                           exchange.getOut().setBody(body);
                        }
                    })
                    .to("activemq:topic:test1");
        }
    }
}
